package com.fyy.kafuka.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @Author fyy
 * @Description
 * @Date 2020/10/22 14:29
 */
public class MyProducer {

    public static void main(String[] args) {
        //producer的配置信息
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"39.101.190.227:9092"); //指定连接集群
        properties.put(ProducerConfig.ACKS_CONFIG,"all"); //ack应答级别
        properties.put(ProducerConfig.RETRIES_CONFIG,"1"); //重试次数
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384); //发送批次大小16kb
        properties.put(ProducerConfig.LINGER_MS_CONFIG,10); //发送等待时间
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432); //缓冲区大小32M
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer"); //key-value的序列化类
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer"); //key-value的序列化类
        //使用自定义分区策略
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.fyy.kafuka.partition.MyPartiton");
        //实例化一个生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        //发送数据(指定分区,指定key)
        producer.send(new ProducerRecord<>("test",0,"test", "ceshi"));
        //带回调的生产者
//        producer.send(new ProducerRecord<>("test", "ceshi"),
//                (metadata, e) -> System.out.println("偏移量："+metadata.offset())
//        );
        //关闭连接
        producer.close();


    }

}
